package com.shujia.streaming

import org.apache.spark.{SparkContext, rdd}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SparkSession}
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}

object Demo5RDDToDStream {
  def main(args: Array[String]): Unit = {
    /**
     * 创建SparkSession
     */
    val sparkSession: SparkSession = SparkSession.builder()
      .master("local[2]")
      .appName("DS2RDD演示")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()

    import sparkSession.implicits._
    import org.apache.spark.sql.functions._

    val sparkContext: SparkContext = sparkSession.sparkContext
    val streamingContext = new StreamingContext(sparkContext, Durations.seconds(5))

    /**
     * 通过spark streaming读取数据得到一个DS
     */
    //hello hello world java hello hadoop
    val linesDS: ReceiverInputDStream[String] = streamingContext.socketTextStream("master", 12345)

    //in_rdd:hello hello world java hello hadoop
    //out_rdd:
    //  hello 3
    //  world 1
    //  java 1
    //  hadoop 1
    val resDS: DStream[(String, Long)] = linesDS.transform((rdd: RDD[String]) => {
      /**
       * rdd处理之后会得到一个新的rdd进行返回
       */
      //      val resultRDD: RDD[(String, Int)] = rdd.flatMap(_.split(" "))
      //        .map((_, 1))
      //        .groupBy(_._1)
      //        .map((kv: (String, Iterable[(String, Int)])) => (kv._1, kv._2.size))
      //      resultRDD

      //def toDF(colNames: String*)
      val linesDF: DataFrame = rdd.toDF("line")
      linesDF.createOrReplaceTempView("words")
      val resRDD: RDD[(String, Long)] = sparkSession.sql(
          """
            |select
            |t1.word as word,
            |count(1) as number
            |from
            |(select
            | explode(split(line,' ')) as word
            |from
            |words) t1
            |group by t1.word
            |""".stripMargin)
        .rdd
        .map((row: Row) => {
          (row.getAs[String](0), row.getAs[Long](1))
        })
      resRDD
    })
    resDS.print()

    streamingContext.start()
    streamingContext.awaitTermination()
    streamingContext.stop()



  }
}
